package com.example.spider.task.limit;

import com.example.spider.task.AbstractSpiderTask;
import com.example.spider.task.SpiderTaskQueue;
import com.example.spider.task.Task;
import org.shoulder.core.concurrent.Threads;

import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 主要用于高限流，如 ip代理爬取：每一段时间发一次请求，不关心响应时间；
 * <p>
 * 对服务器产生固定压力，对不高负载的服务器容易加剧其压力，对空闲的服务器不太容易利用其空闲资源
 *
 * @author lym
 */
public class FixWindowLimit extends AbstractLimitStrategy {

    private static final ConcurrentHashMap<String, AtomicLong> exeEndTime = new ConcurrentHashMap<>(16);

    private static final ConcurrentHashMap<String, AtomicLong> estimateExeTime = new ConcurrentHashMap<>(16);

    public FixWindowLimit(Map<String, Object> map) {
        // only for dejson
        super(map);
    }

    public FixWindowLimit(String key, Duration limitDuration) {
        super(limitDuration);
    }

    public FixWindowLimit(String key, Duration limitDuration, LimitScope limitScope) {
        super(key, limitDuration, limitScope);
    }

    // ===================================================================================

    @Override
    public boolean needLimit(Task<?> task) {
        checkProperties();
        key = genLimitKey(task);

        //return limitDuration != Duration.ZERO;
        // 尝试获取执行乐观锁 //后续加入代理维度，让整体并发量为代理数的倍数  + (task.getActuallyProxy().getIp()
        long now = System.currentTimeMillis();
        long worstExceptedResponseTime = now + 30_000;
        // 拿到其他 spider 设置的执行结束时间
        AtomicLong currentTaskEndTime = new AtomicLong(worstExceptedResponseTime);
        AtomicLong lastEndTime = exeEndTime.computeIfAbsent(key, k -> currentTaskEndTime);

        // 之前没人执行过，成功获取锁
        boolean canExecute = lastEndTime == currentTaskEndTime;
        if (canExecute) {
            return false;
        }
        long lastEnd = lastEndTime.get();
        // 有人执行过，但距离上次执行大于限制 duration 尝试 CAS 获取锁
        if (now - lastEnd > limitDuration.toMillis()) {
            // 注意：这里必须用 new，否则与原来值相等 CAS 必定成功，也不能用 Instant，同一时刻获取的值一样
            boolean swapSuccess = lastEndTime.compareAndSet(lastEnd, worstExceptedResponseTime);
            return !swapSuccess;
        }
        return true;

    }

    @Override
    public void doLimit(AbstractSpiderTask<?> task) {
        // 固定时间间隔放回队列
        long currentTaskExeTime = estimateExeTime.computeIfAbsent(key, k -> new AtomicLong(exeEndTime.get(key).get()))
                .addAndGet(limitDuration.toMillis());
        long waitMills = currentTaskExeTime - System.currentTimeMillis();
        Threads.delay(() -> SpiderTaskQueue.putTask(task), waitMills, TimeUnit.MILLISECONDS);
    }

    @Override
    public <T> void afterRequest(Task<T> task, T result, Throwable ex, long endTime) {
        if (ex == null) {
            AtomicLong lastExeEndTime = exeEndTime.get(key);
            // 最小精度 也用于弥补cpu切换耗时
            final long minPrecision = 1;
            long lastEnd = lastExeEndTime.get();
            if (endTime < lastEnd - minPrecision) {
                lastExeEndTime.compareAndSet(lastEnd, endTime);
            }
        }
    }

    @Override
    public LimitStrategy copyWithNewKey(String s) {
        return new FixWindowLimit(getKey(), getLimitDuration(), getLimitScope());
    }

}
